package br.unifor.cmi.controller.server.impl.jgcs;

import java.io.Serializable;
import java.net.SocketAddress;
import java.util.StringTokenizer;
import java.util.Vector;

import net.sf.jgcs.ControlSession;
import net.sf.jgcs.DataSession;
import net.sf.jgcs.GroupConfiguration;
import net.sf.jgcs.JGCSException;
import net.sf.jgcs.NotJoinedException;
import net.sf.jgcs.Protocol;
import net.sf.jgcs.ProtocolFactory;
import net.sf.jgcs.Service;
import net.sf.jgcs.membership.BlockListener;
import net.sf.jgcs.membership.BlockSession;
import net.sf.jgcs.membership.MembershipListener;
import net.sf.jgcs.membership.MembershipSession;
import net.sf.jgcs.utils.FactoryUtil;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.ow2.cmi.controller.server.impl.jgroups.IIntercepted;

import br.unifor.g2cl.GroupRequest;
import br.unifor.g2cl.IMarshalDataSession;
import br.unifor.g2cl.MarshalDataSessionFactory;
import br.unifor.g2cl.Util;
import br.unifor.g2cl.hl.RpcDispatcher;
import br.unifor.g2cl.hl.StateListener;
import br.unifor.g2cl.hl.StateTransferDataSession;
import br.unifor.util.cluster.jgcs.IDistributedTree;


/**
 * A tree-like structure that is replicated across several members. Updates will be multicast to all group
 * members reliably and in the same order.
 * @author Bela Ban
 * @author <a href="mailto:aolias@yahoo.com">Alfonso Olias-Sanz</a>
 */
public final class SynchronizedDistributedTree implements 
						MembershipListener, 
						IDistributedTree, 
						IIntercepted,
						StateListener,
						BlockListener{

    private volatile Node root=null;
    private final Vector<DistributedTreeListener> listeners=new Vector<DistributedTreeListener>();
    private final Vector<ViewListener> view_listeners=new Vector<ViewListener>();
    private final Vector<SocketAddress> members=new Vector<SocketAddress>();
    
    /** Isso foi adicionado porque no nosso esquema de transfer�cia de estado, ele pode pedir para qualquer membro
     * que possua o estado, j� que em certos casos (spread), n�o haver� um coordenador, logo, n�o temos em quem confiar.
     * Assim, precisamos assegurar que o stateOutgoing retorne NULL quando ele ainda n�o tiver recebido o estado.
     */
    private volatile boolean hasState = false; 
    
    private volatile ControlSession controlSession;
    private volatile MembershipSession membershipSession;
    private volatile DataSession dataSession;
    private volatile StateTransferDataSession stateTransferDataSession;
    private volatile Service service; 
    private volatile Protocol protocol;
    
    
    private MembershipListener membershipListener;
    
    private volatile RpcDispatcher disp=null;
    private volatile String groupname="SynchronizedDistributedTreeGroup";
    private String jgcs_properties = null;
    

    /** Determines when the updates have to be sent across the network, avoids sending unnecessary
     * messages when there are no member in the group */

    // Make this protected so that extensions
    // can control whether or not to send
    private volatile boolean send_message = false;

    private static final Log log=LogFactory.getLog(SynchronizedDistributedTree.class);

    /**
     * A mutex.
     */
    private final Object lock = new Object();
    
    private final Object lockStateTransfer = new Object();


    public SynchronizedDistributedTree() {
    }


    public SynchronizedDistributedTree(final String jgcs_properties) {

        if(jgcs_properties != null) {
            log.info("Loading user defined jgcs properties");
            this.jgcs_properties = jgcs_properties;
        } else {
            log.info("Loading default channel properties");
        }
    }

    public SocketAddress getLocalAddress() {
        return controlSession != null? controlSession.getLocalAddress() : null;
    }
    
    public ControlSession getControlSession() {
    	return controlSession;
    }

	public DataSession getDataSession() {
		return dataSession;
	}
	
	public void setMembershipListener(MembershipListener membershipListener) {
		this.membershipListener = membershipListener; 
	}
    
    //TODO: Verificar dead lock detection
    /*public void setDeadlockDetection(final boolean flag) {
        if(disp != null) {
            disp.setDeadlockDetection(flag);
        }
    }*/

    /* (non-Javadoc)
     * @see org.ow2.cmi.controller.server.impl.jgcs.IDistributedTree#start()
     */
    public void start() throws Exception {
        start(8000);
    }


    public void start(final long timeout) throws Exception {
        
    	if(controlSession != null) {
            return;
        }
    	
    	FactoryUtil jgcsConf = new FactoryUtil(jgcs_properties);
        
        ProtocolFactory o = (ProtocolFactory) jgcsConf.getInstance("jgcsProtocol");
        protocol = o.createProtocol();
        GroupConfiguration group = (GroupConfiguration) jgcsConf.getInstance("jgcsGroup");
        MarshalDataSessionFactory marshalFactory = (MarshalDataSessionFactory) jgcsConf.getInstance("jgcsMarshalDataSession");
        this.service = (Service) jgcsConf.getInstance("jgcsService");
        
        this.groupname = group.toString();
        this.controlSession = protocol.openControlSession(group);
        this.membershipSession = (MembershipSession) controlSession;
        
        if(controlSession instanceof BlockSession){
        	BlockSession blockSession = (BlockSession)controlSession;
        	blockSession.setBlockListener(this);
        }
        
        dataSession = protocol.openDataSession(group); 
        
        IMarshalDataSession marshalDataSession = marshalFactory.getMarshalDataSession(this.controlSession, dataSession);
        
        this.stateTransferDataSession = new StateTransferDataSession(marshalDataSession,
		                								service,
		                								this.membershipSession);
        
        this.stateTransferDataSession.setStateListener(this);
        disp = new RpcDispatcher(stateTransferDataSession, (MembershipSession)controlSession, service, this);
        disp.setMembershipListener(this);
        
        controlSession.join();
         
        log.info("Joined group. My Address: " + getLocalAddress());
        
        //WORK around appia
        Thread.sleep(2000);
        
         
        if (membershipSession.getMembership().getMembershipList().size() == 1) {
            hasState = true;
            log.info("I am the first member. Do not get the group state.");
        } else {
            log.info("Getting the group state");
            this.stateTransferDataSession.requestState(null);
            log.info("Request for the group state sent!");
            synchronized (lockStateTransfer) {
                try {
                    while (!hasState) {
                    	log.info("AGURADANDO...");        	
                    	lockStateTransfer.wait();
                        log.info("ACORDEI!!...");
                    }
                } catch (InterruptedException e) {
                	log.info("VISH>>.");
                    e.printStackTrace();
                }
            }
        }
    }


    public void stop() {
        disp.stop();
        try {
            controlSession.leave();
            stateTransferDataSession.close();
        } catch (JGCSException e) {
            log.error("error closing jgcs connection", e);
        }
        disp = null;
        controlSession = null;
        stateTransferDataSession = null;
        service = null;
        hasState = false;
    } 


    public void addDistributedTreeListener(final IDistributedTree.DistributedTreeListener listener) {
        if(!listeners.contains(listener)) {
            listeners.addElement(listener);
        }
    }


    public void removeDistributedTreeListener(final IDistributedTree.DistributedTreeListener listener) {
        listeners.removeElement(listener);
    }


    public void addViewListener(final ViewListener listener) {
        if(!view_listeners.contains(listener)) {
            view_listeners.addElement(listener);
        }
    }


    public void removeViewListener(final ViewListener listener) {
        view_listeners.removeElement(listener);
    }


    public void add(final String fqn) {
        //Changes done by <aos>
        //if true, propagate action to the group
        if(send_message == true) {
            try {
            	disp.callRemoteMethod(	"_add", 
            							new Class[] { String.class },
            							new Object[] { fqn }, 
            							GroupRequest.GET_ALL, 0); 
            }
            catch(Exception ex) {
                if(log.isErrorEnabled()) {
                    log.error("exception=" + ex);
                }
            }
        }
        else {
            _add(fqn);
        }
    }

    public void add(final String fqn, final Serializable element) {
        add(fqn, element, 0);
    }

    public void reset(final String fqn, final Serializable element)
    {
        reset(fqn, element, 0);
    }

    public void remove(final String fqn) {
        remove(fqn, 0);
    }

    public void add(final String fqn, final Serializable element, final int timeout) {
        //Changes done by <aos>
        //if true, propagate action to the group
        if(send_message == true) {
            try {
            	
            	disp.callRemoteMethod(	"_add", 
            							new Class[] { String.class, Serializable.class },
            							new Object[] { fqn, element }, 
            							GroupRequest.GET_ALL, 
            							timeout);
            }
            catch(Exception ex) {
                if(log.isErrorEnabled()) {
                    log.error("exception=" + ex);
                }
            }
        }
        else {
            _add(fqn, element);
        }
    }

    public void reset(final String fqn, final Serializable element, final int timeout)
    {
        //Changes done by <aos>
        //if true, propagate action to the group
        if(send_message == true) {
            try {
            	
            	disp.callRemoteMethod(	"_reset", 
						new Class[] { String.class, Serializable.class },
						new Object[] { fqn, element }, 
						GroupRequest.GET_ALL, 
						timeout);            	
            }
            catch(Exception ex) {
                if(log.isErrorEnabled()) {
                    log.error("exception=" + ex);
                }
            }
        }
        else {
            _reset(fqn, element);
        }
    }

    public void remove(final String fqn, final int timeout) {
        //Changes done by <aos>
        //if true, propagate action to the group
        if(send_message == true) {
            try {
            	disp.callRemoteMethod(	"_remove", 
						new Class[] { String.class},
						new Object[] { fqn}, 
						GroupRequest.GET_ALL, 
						timeout);   
                
            }
            catch(Exception ex) {
                if(log.isErrorEnabled()) {
                    log.error("exception=" + ex);
                }
            }
        }
        else {
            _remove(fqn);
        }
    }

    public boolean exists(final String fqn) {
        synchronized (lock) {
            return fqn != null && (findNode(fqn) != null);
        }
    }

    public Serializable get(final String fqn) {
        Node n=null;
        if(fqn == null) {
            return null;
        }
        synchronized (lock) {
            n=findNode(fqn);
            if(n != null) {
                return n.element;
            }
        }
        return null;
    }

    public void set(final String fqn, final Serializable element) {
        set(fqn, element, 0);
    }

    public void set(final String fqn, final Serializable element, final int timeout) {
        //Changes done by <aos>
        //if true, propagate action to the group
        if(send_message == true) {
            try {
            	
            	disp.callRemoteMethod(	"_set", 
						new Class[] { String.class, Serializable.class },
						new Object[] { fqn, element }, 
						GroupRequest.GET_ALL, 
						timeout); 
            }
            catch(Exception ex) {
                if(log.isErrorEnabled()) {
                    log.error("exception=" + ex);
                }
            }
        }
        else {
            _set(fqn, element);
        }
    }


    /** Returns all children of a Node as strings */
    public Vector<String> getChildrenNames(final String fqn) {
        Vector<String> ret=new Vector<String>();
        Node n;

        if(fqn == null) {
            return ret;
        }
        synchronized (lock) {
            n=findNode(fqn);
            if(n == null || n.children == null) {
                return ret;
            }
            for(Node child : n.children) {
                ret.addElement(child.name);
            }
        }
        return ret;
    }


    public String print() {
        StringBuilder sb=new StringBuilder();
        int indent=0;

        synchronized (lock) {
            if(root == null) {
                return "/";
            }

            sb.append(root.print(indent));
        }
        return sb.toString();
    }


    /**
     * Returns the name of the group that the DistributedTree is connected to
     * @return String
     */
    public String  getGroupName()           {return groupname;}

    /**
     * Returns the number of current members joined to the group
     * @return int
     */
    public int getGroupMembersNumber()			{return members.size();}




    /*--------------------- Callbacks --------------------------*/

    public void _add(final String fqn) {
        _add(fqn, null);
    }


    public void _add(final String fqn, final Serializable element) {
        Node curr, n;
        StringTokenizer tok;
        String child_name;
        String tmp_fqn="";

        synchronized (lock) {
            if(root == null) {
                root=new Node("/", null);
                notifyNodeAdded("/", null);
            }
            if(fqn == null) {
                return;
            }
            curr=root;
            tok=new StringTokenizer(fqn, "/");

            while(tok.hasMoreTokens()) {
                child_name=tok.nextToken();
                tmp_fqn=tmp_fqn + '/' + child_name;
                n=curr.findChild(child_name);
                if(n == null) {
                    n=new Node(child_name, null);
                    curr.addChild(n);
                    if(!tok.hasMoreTokens()) {
                        n.element=element;
                        notifyNodeAdded(tmp_fqn, element);
                        return;
                    } else {
                        notifyNodeAdded(tmp_fqn, null);
                    }
                }
                curr=n;
            }
            // If the element is not null, we install it and notify the
            // listener app that the node is modified.
            if(curr.element == null && element != null){
                curr.element=element;
                notifyNodeModified(fqn, null, element);
            } else if(curr.element != null) {
                log.debug("Cannot add " + element + " at "
                        + tmp_fqn + " because " + curr.element + " exists");
            }
        }
    }


    public void _remove(final String fqn) {
        Node curr, n;
        StringTokenizer tok;
        String child_name=null;

        synchronized (lock) {
            if(fqn == null || root == null) {
                return;
            }
            curr=root;
            tok=new StringTokenizer(fqn, "/");

            while(tok.countTokens() > 1) {
                child_name=tok.nextToken();
                n=curr.findChild(child_name);
                if(n == null) {
                    return;
                }
                curr=n;
            }
            try {
                child_name=tok.nextToken();
                if(child_name != null) {
                    n=curr.removeChild(child_name);
                    if(n != null) {
                        notifyNodeRemoved(fqn);
                    }
                }
            }
                catch(Exception ex) {
            }
        }
    }


    public void _set(final String fqn, final Serializable element) {
        Node n;
        Serializable old_el=null;

        synchronized (lock) {
            if(fqn == null || element == null) {
                return;
            }
            n=findNode(fqn);
            if(n == null) {
                if(log.isErrorEnabled()) {
                    log.error("node " + fqn + " not found");
                }
                return;
            }
            old_el=n.element;
            n.element=element;
            notifyNodeModified(fqn, old_el, element);
        }
    }

    /** similar to set, but does not error if node does not exist, but rather does an add instead */
    public void _reset(final String fqn, final Serializable element) {
        Node n;
        Serializable old_el=null;

        synchronized (lock) {
            if(fqn == null || element == null) {
                return;
            }
            n=findNode(fqn);
            if(n == null) {
                _add(fqn, element);
            }
            else {
                old_el=n.element;
                n.element=element;
            }
            notifyNodeModified(fqn, old_el, element);
        }
    }

    /*----------------- End of  Callbacks ----------------------*/


    /*-------------------- State Exchange ----------------------*/

    public void stateIncoming(byte[] state) {
    	log.info("RECEBENDO ESTADO!!!");
    	
        Object new_state = null;
 
        try {
        	if(state != null){
        		log.info("Incoming State Length: "+ state.length);
        		log.info("State: "+ state.hashCode());
        		new_state=Util.getObjectFromByte(state,0, state.length);
        		
        	}            
        }
        catch(Throwable ex) {
            if(log.isErrorEnabled()) {
            	ex.printStackTrace();
                log.error("exception unmarshalling state: " + ex);
            }
            return;
        }
        if(new_state == null) {
            return;
        }
        if(!(new_state instanceof Node)) {
            if(log.isErrorEnabled()) {
                log.error("object is not of type 'Node'");
            }
            return;
        }
        root=((Node)new_state).copy();
        hasState = true;
        // State transfer needs to notify listeners in the new
        // cluster member about everything that exists.  This
        // is working ok now.
        this.notifyAllNodesCreated(root, "");
 
        log.info("ACORDANDO!!!");
        synchronized (lockStateTransfer) {
        	lockStateTransfer.notify();
        }
    }
 
 
    public byte[] stateOutgoing() {
    	log.info(this.getClass().getCanonicalName() + " -> " + "FORNECENDO ESTADO");
    	if (!hasState) {
    		log.info("STATE NULL!!");
            return null;
        }
        Object copy = (root != null) ? root.copy()
                                     : null;
        try {
        	log.info("OBJ = "+copy);
            byte[] state = Util.getArrayFromObject(copy);
            
            log.info("State: "+ state.hashCode());
            log.info("Outcoming State Length: "+ state.length);
            
            return state;
        } catch (Throwable ex) {
            if (log.isErrorEnabled()) {
                log.error("exception marshalling state: " + ex);
            }
            return null;
        }
    } 



    /*------------------- Membership Changes ----------------------*/

	public void onExcluded() {
        if(membershipListener != null){
        	membershipListener.onMembershipChange();
        }
	}


    public void onMembershipChange() {
        try {
            Vector<SocketAddress> new_mbrs = new Vector<SocketAddress>(membershipSession.getMembership().getMembershipList());
            log.info("VIEW: "+new_mbrs);
             
            if(new_mbrs != null) {
                sendViewChangeNotifications(new_mbrs, members); // notifies observers (joined, left)
                members.removeAllElements();
                for(SocketAddress mbr : new_mbrs) {
                    members.addElement(mbr);
                }
            }
            //if size is bigger than one, there are more peers in the group
            //otherwise there is only one server.
            send_message=true;
            send_message=members.size() > 1;
            //System.out.println("SEND MESSAGE: " + send_message);
        } catch (NotJoinedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        if(membershipListener != null){
        	membershipListener.onMembershipChange();
        }
    }


    void sendViewChangeNotifications(final Vector<SocketAddress> new_mbrs, final Vector<SocketAddress> old_mbrs) {
        Vector<SocketAddress> joined, left;

        if(view_listeners.isEmpty() || old_mbrs == null || new_mbrs == null) {
            return;
        }


        // 1. Compute set of members that joined: all that are in new_mbrs, but not in old_mbrs
        joined=new Vector<SocketAddress>();
        for(SocketAddress mbr : new_mbrs) {
            if(!old_mbrs.contains(mbr)) {
                joined.addElement(mbr);
            }
        }


        // 2. Compute set of members that left: all that were in old_mbrs, but not in new_mbrs
        left=new Vector<SocketAddress>();
        for(SocketAddress mbr : old_mbrs) {
            if(!new_mbrs.contains(mbr)) {
                left.addElement(mbr);
            }
        }
        notifyViewChange(joined, left);
    }


    private Node findNode(final String fqn) {
        Node curr=root;
        StringTokenizer tok;
        String child_name;

        if(fqn == null || root == null) {
            return null;
        }
        if("/".equals(fqn) || "".equals(fqn)) {
            return root;
        }

        tok=new StringTokenizer(fqn, "/");
        while(tok.hasMoreTokens()) {
            child_name=tok.nextToken();
            curr=curr.findChild(child_name);
            if(curr == null) {
                return null;
            }
        }
        return curr;
    }


    void notifyNodeAdded(final String fqn, final Serializable element) {
        for(DistributedTreeListener distributedTreeListener : listeners) {
            distributedTreeListener.nodeAdded(fqn, element);
        }
    }

    void notifyNodeRemoved(final String fqn) {
        for(DistributedTreeListener distributedTreeListener : listeners) {
            distributedTreeListener.nodeRemoved(fqn);
        }
    }

    void notifyNodeModified(final String fqn, final Serializable old_element, final Serializable new_element) {
        for(DistributedTreeListener distributedTreeListener : listeners) {
            distributedTreeListener.nodeModified(fqn, old_element, new_element);
        }
    }

    /** Generates NodeAdded notifications for all nodes of the tree. This is called whenever the tree is
     initially retrieved (state transfer) */
    void notifyAllNodesCreated(final Node curr, final String tmp_fqn) {
        // We need a local string here to handle the empty string (root)
        // otherwise, we start off with two slashes in the path.
        String path = "";
        if(curr == null) {
            return;
        }
        if(curr.name == null) {
            if(log.isErrorEnabled()) {
                log.error("curr.name is null");
            }
            return;
        }
        // If we're the root node, then we supply a "starter" slash.
        // This lets us properly initiate the recursion with an empty
        // string, and then prepend a slash for each additional depth
        path = (curr.equals(root)) ? "/" : tmp_fqn;

        // Recursion must occur _before_ we look for children, or we
        // never notifyNodeAdded() for leaf nodes.
        notifyNodeAdded(path, curr.element);
        if(curr.children != null) {
            for(Node n : curr.children) {
                log.debug("*** nodeCreated(): tmp_fqn is " + tmp_fqn);
                notifyAllNodesCreated(n, tmp_fqn + '/' + n.name);
            }
        }
    }


    void notifyViewChange(final Vector<SocketAddress> new_mbrs, final Vector<SocketAddress> old_mbrs) {
        for(ViewListener viewListener : view_listeners) {
            viewListener.viewChange(new_mbrs, old_mbrs);
        }
    }


    private static class Node implements Serializable {
        final String name;
        volatile Vector<Node> children=null;
        volatile Serializable element=null;
        private static final long serialVersionUID=-635336369135391033L;


        Node(final String name, final Serializable element) {
            this.name =name;
            this.element = element;
        }

        void addChild(final Node n) {
            if(n == null) {
                return;
            }
            if(children == null) {
                children = new Vector<Node>();
            }
            if(!children.contains(n)) {
                children.addElement(n);
            }
        }


        Node removeChild(final String rel_name) {
            Node n=findChild(rel_name);

            if(n != null) {
                children.removeElement(n);
            }
            return n;
        }


        Node findChild(final String relative_name) {
            if(children == null || relative_name == null) {
                return null;
            }
            for(Node child : children) {
                if(child.name == null) {
                    if(log.isErrorEnabled()) {
                        log.error("child.name is null for " + relative_name);
                    }
                    continue;
                }

                if(child.name.equals(relative_name)) {
                    return child;
                }
            }

            return null;
        }

        @Override
        public boolean equals(final Object other) {
            return other != null
                && other instanceof Node
                && ((Node) other).name != null
                && name != null
                && name.equals(((Node) other).name);
        }

        @Override
        public int hashCode() {
            return name.hashCode();
        }


        @SuppressWarnings("unchecked")
        Node copy() {
            Node ret=new Node(name, element);

            if(children != null) {
                ret.children= (Vector<Node>) children.clone();
            }
            return ret;
        }


        String print(int indent) {
            StringBuilder sb=new StringBuilder();
            boolean is_root=name != null && "/".equals(name);

            for(int i=0; i < indent; i++) {
                sb.append(' ');
            }
            if(!is_root) {
                if(name == null) {
                    sb.append("/<unnamed>");
                } else {
                    sb.append('/' + name);
                    // if(element != null) sb.append(" --> " + element);
                }
            }
            sb.append('\n');
            if(children != null) {
                if(is_root) {
                    indent=0;
                } else {
                    indent+=4;
                }
                for(Node child : children) {
                    sb.append(child.print(indent));
                }
            }
            return sb.toString();
        }

        @Override
        public String toString() {
            if(element != null) {
                return "[name: " + name + ", element: " + element + ']';
            } else {
                return "[name: " + name + ']';
            }
        }

    }


	public boolean hasState() {
		return hasState;
	}


	public Protocol getProtocol() {
		return protocol;
	}

	public Service getService() {
		return service;
	}
	
	public void onBlock() {
		
		BlockSession blockSession = (BlockSession)controlSession;
		try {
			blockSession.blockOk();
		} catch (NotJoinedException e) {
			e.printStackTrace();
		} catch (JGCSException e) {
			e.printStackTrace();
		}
		
	}

}